package com.weiss.jpro.service.impl;

import cn.hutool.Hutool;
import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.weiss.jpro.dto.Result;
import com.weiss.jpro.entity.SeckillVoucher;
import com.weiss.jpro.entity.VoucherOrder;
import com.weiss.jpro.mapper.VoucherOrderMapper;
import com.weiss.jpro.service.ISeckillVoucherService;
import com.weiss.jpro.service.IVoucherOrderService;
import com.weiss.jpro.utils.RedisIdWorker;
import com.weiss.jpro.utils.SimpleRedisLock;
import com.weiss.jpro.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    // 超卖问题
    // 超卖问题是典型的多线程安全问题，针对这一问题常见的解决方案就是加锁：
    // 悲观锁：
    // 认为线程安全问题一定会发生，因此在操作数据之前先获取锁，确保线程串行执行。
    // 如Synchronized、Lock都属于悲观锁，数据库中互斥的锁也属于悲观锁
    // 乐观锁：
    // 认为线程安全问题不一定会发生，因此不加锁，只在更新数据时去判断有没有其他线程对数据做了修改
    // 如果没有修改则认为是安全的，自己才更新数据
    // 如果已经被其他线程修改说明发生了安全问题，此时可以重试或异常
    // 比如我现在查到了一个库存，查到库存之后要去更新库存，在更新库存之前判断以下有没有其他人修改库存
    // 看一看，对比对比，如果改了、和之前查的不一样了，就说明有人改了，存在安全问题，就重试或者抛异常
    // 如果和之前查的一样，数据没有变化，线程安全，则可以方向修改
    //
    // 因为乐观锁这种方案不用去加锁，而是在执行时去进行判断，因此他的性能要比悲观锁好很多
    // 乐观锁的关键点在于更新的时候怎么知道别人有没有来修改过，常见的方式有两种：
    //
    // 版本号法(应用最广泛、最普遍)：
    // 给数据加上一个版本号字段，在多线程并发时根据版本号去判断有没有发生修改
    // 每当数据做一次修改，版本号就会加一
    // 线程1查询库存和版本号：stock=1，version=1，线程2查询库存和版本号：stock=1，version=1
    // 线程1判断库存是否大于0，扣减库存，set stock=stock-1，version=version+1 where id =xx and version=1，执行成功(此时version=1，执行完成version=2)
    // 线程2判断库存是否大于0，扣减库存，set stock=stock-1，version=version+1 where id =xx and version=1， 执行失败(此时version=2)
    //
    // 方法2，在版本号基础上做了一些简化
    // CAS法，Compare-And-Swap
    // 版本号法是用版本标识数据有没有变化，查询时要把版本查出来，每次更新数据时版本也要跟着更新
    // 既然每次都要更新版本，查到的版本和更新的版本一致则证明查询到修改之间没有发生更新
    // 当前业务查版本version时也会查询库存stock，更新时库存也会更新，stock和version所做的事情是一样的，所以可以用stock来代替version
    // 查询的时候把库存stock查出来，更新的时候判断stock和查出的stock是不是一样的，如果一样也可以证明没有人进行修改
    // 则可以去掉version字段，用数据本身是否发生变化去判断线程是否安全
    // 线程1查询库存，stock=1，线程2查询库存stock=1
    // 线程1判断库存是否大于0，扣减库存，set stock=stock-1 where id=xx and stock=1，执行成功(此时stock=1，执行完毕stock=0)
    // 线程2判断库存是否大于0，扣减库存，set stock=stock-1 where id=xx and stock=1，执行失败(此时stock=0)


    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;;
    
    @Resource
    private StringRedisTemplate stringRedisTemplate; // 获取锁对象需要

    @Resource
    private RedissonClient redissonClient; // 通过redisson获取锁

    // 改造后流程梳理：
    // 调用优惠券抢购首先会进入seckillVoucher方法，
    // 进入后会先去调用lua脚本，尝试判断库存是否充足，用户有没有购买资格，
    // 根据脚本返回值0，1，2判断有没有购买资格，没有资格直接返回给前端信息
    // 有资格不是直接下单，而是将订单创建出来，放到阻塞队列里面去
    //
    // 放到阻塞队列之后，要去执行异步下单，我们认为应该能够100%成功，因为lua脚本已经进行判断过了，
    // 我们利用阻塞队列，线程池，在类初始化的时候去执行线程池，线程池的业务就是从阻塞队列中不断取出订单然后去创建，
    // 通过handlerVoucherOrder进行订单创建，同样先获取锁(冗余，为了兜底)，拿到锁后去通过代理对象开启事务调用crtVoucherOrder方法，
    // crtVoucherOrder同样进行库存和秒杀资格判断(冗余)，然后保存订单到数据库

    // 秒杀优化的思路是什么？
    // 修改同步下单为异步下单
    // 同步下单：来了判断有没有资格，有资格就立即去下单，因为下单，扣库存等要加各种各样的事务和锁
    // ，就导致了一系列的业务全部执行完就会耗时非常久，最低耗时都能达到一百多毫秒甚至两百多毫秒，平均耗时达到了四百多毫秒，
    // 异步下单：将业务分为两部分，一部分是对于资格的判断，有资格就立即结束返回订单id，至于耗时久的下单业务可以异步去完成
    // 1 先利用redis完成库存余量、一人一单判断，完成抢单业务
    // 2 再将下单业务放入阻塞队列，利用独立线程异步下单
    // 基于阻塞队列的异步秒杀存在哪些问题？
    // 我们使用的是jdk里面的阻塞队列，而这个阻塞队列使用的是jvm的内存，如果不加以限制，
    // 在高并发的情况下就会有无数的订单对象需要去创建并且放到阻塞队列里，可能会导致将来内存溢出，
    // 所以在创建阻塞队列的时候设置了一个队列的长度(1024*1024)，有一个上限，但是如果队列存满了，再有新的订单就无法添加进去，
    // 现在我们是基于内存来保存这些订单信息的，如果服务突然宕机，内存里的所有订单信息都会丢失，这样就出现了数据不一致的情况，
    // 另一种情况是现在有一个线程从队列里取出了一个下单的任务要去执行，就在此时发生了严重的事故，比如出现了什么异常，这样任务就没有执行，
    // 但是取出队列任务就没有了，以后就再也不会执行了，就出现了任务丢失的情况，再次出现了数据不一致，这两种情况都会导致数据安全的问题
    // 1 内存限制问题
    // 2 数据安全问题

    // redis消息队列实现异步秒杀
    // 消息队列（message queue），字面意思就是存放消息的队列，最简单的消息队列包括三个角色：
    // 消息队列：存储和管理消息，也被称为消息代理（message broker）
    // 生产者：发送消息到消息队列
    // 消费者：从消息队列获取消息并处理消息
    // 解除耦合，提高工作效率（快递柜）
    //
    // 消息队列与之前阻塞队列有两点不同：
    // 1 消息队列是在jvm以外的一个独立服务，它不受jvm内存的限制，解决了阻塞队列内存溢出的问题
    // 2 消息队列不仅仅是做数据存储，还要确保数据安全，存进消息队列的所有消息要做持久化，
    // 不管服务宕机还是重启消息都不会丢失，消息队列还要在消息传递给消费者以后，要求消费者做消息的确认，
    // 如果消费者没有确认，消息仍然会存在消息队列中，下一次会再投递给消费者让他继续处理，直到消费者确认为止，确保消息至少消费一次
    //
    // redis提供了三种不同的方式来实现消息队列：
    // list结构：基于list结构模拟消息队列
    // PubSub：基本的点对点消息模型
    // Stream：比较完善的消息队列模型
    //
    // 基于list结构模拟消息队列
    // redis的list结构是一个双向链表，很容易模拟出队列效果
    // 利用lpush结合rpop、或者rpush结合lpop来实现
    // 当队列中没有消息时rpop或lpop操作会返回null，并不想jvm的阻塞队列那样会阻塞并等待消息，
    // 因此应该使用brpop或者blpop来实现阻塞效果
    // 优点：
    // 1 利用redis存储，不受jvm内存上限
    // 2 基于redis持久化，数据安全性有保证
    // 3 可以满足消息有序性
    // 缺点：
    // 1 无法避免消息丢失
    // 2 只支持单消费者
    // （rabbitmq中队列的消息也只可以转发给一个消费者，但是可以通过交换机把一条消息转发给多个队列，实现一条消息被多个消费者接收处理的效果）
    //
    // 基于PubSub的消息队列
    // PubSub(发布订阅)是redis2.0版本引入的消息传递模型，消费者可以订阅一个或多个channel，
    // 生产者向对应channel发消息后，所有订阅者都能收到相关消息
    // SUBSCRIBE channel [channel] 订阅一个或多个频道
    // PUBLISH channel msg 向一个频道发送消息
    // PSUBSCRIBE pattern[pattern] 订阅与pattern（通配符）格式匹配的所有频道（？单个 *多个 [ae]a或者e）
    // 例：生产者 publish order.queue msg1 -> message queue -> 消费者1 subscribe order.que 消费者2 psubscribe order.*
    // 优点：
    // 采用发布订阅模型。支持多生产、多消费
    // 缺点：
    // 1 不支持数据持久化 （redis基本数据类型支持持久化）
    // 2 无法避免消息丢失 （发布消息时，没有消费者监听就会丢失）
    // 3 消息堆积有上限、超出时消息丢失（发送一个消息时，如果有消费者监听，会在消费者那里有一个缓存区域，把消息缓存下来，
    // 接下来消费者去处理，如果消费者处理比较慢，在处理过程中又来了十几条消息，这些消息都会缓存在客户端，也就是消费者那里，
    // 而消费者缓存的空间是有上限的，如果超出就会丢失）
    //
    // 基于Stream的消息队列
    // 单消费模式
    // stream是redis5.0引入的一种新数据类型（数据类型，和string、hash一样，支持数据持久化），可以实现一个功能非常完善的消息队列
    // 发送消息：XADD key [NOMKSTREAM]（如果队列不存在，是否自动创建队列，默认自动创建）
    // [MAXLEN|MINID [=|~] threashold [LIMIT count]]（设置消息队列的最大消息数量）
    // *|ID （消息的唯一id，*代表有redis自动生成，格式是"时间戳-递增数字"，例如"1644804662707-0"）
    // field value [field value ...](发送到消息队列中的消息，称为entry，格式就是多个key-value键值对)
    // 例：创建名为users的队列，并向其中发送一个消息，内容是：{name=javk, age=21}，并且使用redis自动生成id
    // XADD users * name jack age 21
    // XLEN users 查看消息队列中的消息数量
    // 读取消息：XREAD [COUNT count](每次读取消息的最大数量) [BLOCK milliseconds](当没有消息时，是否阻塞，阻塞时长)
    // STREAMS key [key ...](要从哪个队列读取消息，key就是队列名) ID [ID ...](起始id，只返回大于该id的消息，0：代表从第一个消息开始，$：代表从最新的消息开始)
    // 例：XREAD COUNT 1 STREAMS s1 0 可以重复读取
    // XREAD COUNT 1 BLOCK 0 STREAMS s1 $ 永久阻塞
    // 在业务开发中可以利用循环调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下：
    // while(true) {
    //    // 尝试获取队列中的消息，阻塞2秒
    //    Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $");
    //    if (msg == null){
    //        continue;
    //    }
    //    // 处理消息
    //    handleMessage(msg);
    // }
    // 当我们指定其实id为$时，代表读取最新消息，如果我们处理一条消息的过程中，
    // 又有超过1条以上的消息达到队列，则下次只能获取到最新的一条，会出现漏读消息的问题
    // stream类型消息队列XREAD命令特点：
    // 消息可回溯（永久保存）
    // 一个消息可以被多个消费者读取
    // 可以阻塞读取
    // 有消息漏读的风险
    //
    // 消息队列-消费者组
    // 消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列，具备以下特点：
    // 1 消息分流：队列中的消息会分流给组内不同的消费者，而不是重复消费，从而加快消息处理的速度
    // 我们的多个消费者只要在一个组里，他们之间就是一个竞争关系，这样处理消息的速度就大大加快了， （从一定程度上避免消息堆积的问题）
    // 2 消息标识：消费者会维护一个标识，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标识之后读取数据，确保每一个消息都会被消费，（避免漏读）
    // 3 消息确认：消费者获取消息后，消息处于pending状态，并存入一个pending-list，当处理完成后需要通过XACK来确认消息，
    // 标记消息为已处理，才会从pending-list移除，（可以解决消息丢失的问题）
    //
    // 创建消费者组：XGROUP CREATE key groupName ID [MKSTREAM]
    // key 队列名称 groupName 消费者组名称 ID 起始ID标识，$代表队列中最后一个消息，0代表队列中第一个消息
    // 如果我们创建组的时候，队列中没有消息或者队列中已经有很多消息了，而那些消息又不想消费，可以选择$
    // 如果说队列中消息还想要重新消费一下，可以选择0
    // MKSTREAM 队列不存在时自动创建队列
    //
    // 删除指定的消费者组 XGROUP DESTROY key groupName
    // 给指定消费者组添加消费者 XGROUP CREATECONSUMER key groupname consumername
    // 删除消费者组中的指定消费者 XGROUP DELCONSUMER key groupname consumername
    // 一般情况下并不需要自己去添加消费者，当我们从消费者组指定一个消费者去监听消息时，如果不存在会自动创建
    //
    // 从消费者组读取消息：XREADGRUOP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key...] ID [ID...]
    // group 消费者组名称
    // consumer 消费者名称，如果消费者不存在，会自动创建一个消费者
    // count 本次查询的最大数量
    // BLOCK milliseconds 当没有消息时最长等待时间
    // NACK 无需手动确认ACK，获取消息后自动确认
    // STREAMS key 指定队列名称
    // ID 获取消息起始的ID：">" 从下一个未消费的消息开始，其他：根据指定pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始
    //
    // 确认消息 XACK key group ID [ID...]
    // 查看pending-list XPENDING key group - + COUNT
    //
    // 当消费者进入组里后，所有消息存入此消费者的pending-list中，处于pending状态,
    // 只有当我们手动xack来确定消息，才会从pending-list中移除，此时可以避免消息读取后，
    // 结果程序宕机（不是redis），我们程序还没来得及处理,那么就可以从pending-list中重新获取
    //
    // 例：
    // XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
    // XACK s1 g1 1646339018049-0
    // XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
    // 未确认，出现异常
    // XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0 读取pending-list中的第一条消息
    // XPENDING s1 g1 - + 10 返回pending-list中所有未确认消息
    // XACK s1 g1 164xxxxxxxxxx-0 pending-list中未确认消息的id
    //
    // 消费者监听消息的基本思路：
    // while(true){
    //    // 尝试监听队列，使用阻塞模式，最长等待2000毫秒
    //    Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >");
    //    if (msg == null) { // null说明没有消息，继续下一次
    //        continue;
    //    }
    //    try {
    //        // 处理消息完成后一定要ack
    //        handleMessage(msg);
    //    } catch (Exception e) {
    //        while(true){
    //            Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0");
    //            if (msg == null) { // null说明没有异常消息，所有消息都已确认，结束循环
    //                break;
    //            }
    //            try {
    //                // 说明有异常消息，再次处理
    //                handleMessage(msg);
    //            } catch (Exception ex) {
    //                // 再次处理异常，记录日志，继续循环
    //                continue;
    //            }
    //        }
    //    }
    // }
    //
    // stream类型消息队列的XREADGROUP命令特点：
    // 消息可回溯
    // 可以多消费者争抢消息，加快消费速度
    // 可以阻塞读取
    // 没有消息漏读的风险
    // 有消息确认机制，保证消息至少被消费一次
    //
    // redis消息队列
    // list 消息持久化 支持 阻塞读取 支持 消息堆积处理 受限于内存空间，可以利用多消费者加快处理 消息确认机制 不支持 消息回溯 不支持
    // pubsub 消息持久化 不支持 阻塞读取 支持 消息堆积处理 受限于消费者缓冲区 消息确认机制 不支持 消息回溯 不支持
    // stream 消息持久化 支持 阻塞读取 支持 消息堆积处理 受限于队列长度，可以利用消费者组提高消费速度，减少堆积 消息确认机制 支持 消息回溯 支持
    //
    // 如果对消息的限制更加严格，应该使用更加专业的消息队列，如rabbitMQ，rocketMQ等等
    // 这是因为stream虽然支持消息的持久化，这种持久化是依赖于redis本身持久化的，
    // 而redis的持久化也是有丢失风险的，而且消息确认机制只支持消费者的确认机制，
    // 而不支持生产者的消息确认机制，如果生产者在发消息的过程中丢失就无法处理，
    // 另外还有消息的事务机制、在多消费下的消息有序性等等，这些问题都需要更加强大的消息队列去支持，
    // 如果是一些中小型企业，对于消息队列的要求并没有那么多，stream已经可以满足我们的业务需要了

    // 基于redis的stream结构作为消息队列，实现异步秒杀下单
    // 1 创建一个stream类型的消息队列，名为stream.orders
    // 直接在redis里面创建 XGROUP CREATE stream.orders g1 0 MKSTREAM
    // 2 修改之前秒杀的lua脚本，在确认有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId，userId，orderId
    // 使用redis实现消息队列，发消息就是一个redis的命令，可以直接写在lua脚本里面，就可以减少一次java与redis之间的交互，效率会更高
    // 3 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

    // 使用的是静态常量和静态代码块，这个类一加载，脚本就初始化完成了
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    // 由于UNLOCK_SCRIPT是静态的，在静态代码块进行初始化
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        // 设置脚本位置
        SECKILL_SCRIPT.setLocation(new ClassPathResource("mapper/stream-seckill.lua")); // 使用spring提供的ClassPathResource寻找classpath下的资源
        SECKILL_SCRIPT.setResultType(Long.class); // 返回值类型
    }

    // 阻塞队列特点：当一个线程尝试从队列里获取元素时，如果队列没有元素，这个线程就会被阻塞，直到队列里面有元素为止
    // 已经不需要阻塞队列了
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024); // 基于数组实现的阻塞队列
    // 创建线程池用于从阻塞队列读取 (ctrl+shift+u改为大写)
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    // 线程任务：应该在用户抢购秒杀之前开始执行，因为用户一旦开始秒杀，就会向阻塞队列不断写入订单，任务就应该取出订单信息，
    // 项目只要一启动，用户随时都有可能来进行抢购，所以该任务应该在类进行初始化之后就赶紧执行，
    // 通过spring的@PostConstruct注解，该注解代表在当前类初始化完毕后立即执行
    // VoucherOrderServiceImpl一旦初始化完毕，init方法就会执行，线程任务被提交，VoucherOrderHandler里的run方法就会执行
    // (静态代码块无法使用成员内部类，要用也得用实例代码块，只不过实例代码块在构造函数之前就执行，那时该实例的注入的资源就不能用)

    @PostConstruct
    private void init() {
        // 提交线程任务
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }
    // 线程任务
    // 线程任务也因该修改为从消息队列中获取消息
    private class VoucherOrderHandler implements Runnable {

        // 代码逻辑
        // 上来先尝试从消息队列中读取消息，如果没有消息，continue，直接结束进入下一轮循环
        // 如果有消息，则代表需要处理订单，从消息队列中取出消息，解析订单信息，完成下单，进行ACK确认
        // 如果处理消息过程产生异常，导致没有进行ACK确认，则进入pending-list中处理异常消息
        // 处理完成后，继续读取pending-list，没有异常消息后，break
        // 如果pending-list处理过程中继续抛出异常，通过线程休眠防止频繁处理，
        // 然后进入下一轮循环继续处理，直到pending-list中的所有异常全部处理完为止，确保异常订单一定能够得到处理
        @Override
        public void run() {
            // 利用循环不断从阻塞队列里取
            while (true) {
                // ctrl+alt+t快速抛出异常
                try {
                    // 获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    // c1 消费者名字将来应该配到yaml文件里，不同的节点(启动多节点的话)，消费者名字应该有多个，这里直接写死为c1
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), // ctrl+p 查看参数
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed()) // ctrl+鼠标左键 步入
                    );
                    // 判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        // 如果获取失败，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 获取成功，可以下单
                    // 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0); // <消息id, key, value>
                    Map<Object, Object> values = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 创建订单 alt+enter创建handleVoucherOder方法到ServiceImpl当中
                    handleVoucherOrder(voucherOrder);
                    // 进行ACK确认 XACK stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                } catch (Exception e) {
                    // 通过日志记录异常
                    log.error("处理订单异常", e);
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                // ctrl+alt+t快速抛出异常
                try {
                    // 获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1), // ctrl+p 查看参数
                            StreamOffset.create("stream.orders", ReadOffset.from("0")) // ctrl+鼠标左键 步入
                    );
                    // 判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        // 如果获取失败，说明pending-list中没有异常消息，结束循环
                        break;
                    }
                    // 获取成功，可以下单
                    // 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0); // <消息id, key, value>
                    Map<Object, Object> values = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 创建订单 alt+enter创建handleVoucherOder方法到ServiceImpl当中
                    handleVoucherOrder(voucherOrder);
                    // 进行ACK确认 XACK stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                } catch (Exception e) {
                    // 通过日志记录异常
                    log.error("处理pending-list订单异常", e);
                    try {
                        Thread.sleep(20);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
        }
    }
    /***
    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            // 利用循环不断从阻塞队列里取
            while (true) {
                // ctrl+alt+t快速抛出异常
                try {
                    // 获取队列中的订单信息
                    // take()方法是一个阻塞方法，获取队列的头部，如果有需要则等待直到元素可用
                    // 所以不需要担心死循环对cpu造成影响，如果没有元素会卡在take
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 创建订单 alt+enter创建handleVoucherOder方法到ServiceImpl当中
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    // 通过日志记录异常
                    log.error("处理订单异常", e);
                }
            }
        }
    ***/
    }

    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 创建锁对象需要用户id,这里获取用户id不能从userHolder里面获取了，
        // 因为现在是多线程，是从线程池里获取的一个全新的线程，不是主线程去做的，
        // 所以从threadlocal里是取不到用户信息的，这里只能从voucherOder里取用户id
        // 获取用户
        Long userId = voucherOrder.getUserId();

        // 创建锁对象
        RLock lock1 = redissonClient.getLock("lock:order:" + userId);

        // 获取锁
        boolean isLock = lock1.tryLock();
        // 判断是否获取锁成功
        if (!isLock) {
            // 获取失败，这里是异步处理，已经不需要返回给前端信息了
            // 而且在这里理论上不需要加锁也ok，因为已经在redis里做了并发判断了
            // 在这里加锁是为了兜底，以防万一，防止redis没有判断成功
            // 记录日志
            log.error("一个人只允许下一单");
            return;
        }
        // ctrl+alt+t 快速抛出异常
        try {
            // 获取代理对象(事务)
            // 这里同样拿不到voucherId，因为currentProxy底层也是通过threadLocal获取的
            // 现在想要proxy去触发事务，需要将事务对象在主线程中提前获取
            // IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            proxy.crtVoucherOrder(voucherOrder);
        } finally {
            // 释放锁
            lock1.unlock();
        }
    }

    // 创建代理对象(事务)的成员变量，方便后续子线程获取
    // 相当于把代理对象放到当前类里面了，从子线程去取的时候其实就是拿到一个线程的代理对象，不用再取临时获取了
    private IVoucherOrderService proxy;

    @Override
    public Result seckillVoucher(Long voucherId) {
        // 获取用户
        Long userId = UserHolder.getUser().getId();
        // 订单id
        long orderId = redisIdWorker.nextId("order");
        // 1. 执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), // 传入空集合
                voucherId.toString(), userId.toString(), String.valueOf(orderId) // 基本类型orderId，另外两个为包装类型
        );
        int r = result.intValue(); // long型转为int

        // 2. 判断结果是否为0
        if (r != 0) {
            // 2.1 不为0，没有购买资格，1，2
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }

        // 将创建订单对象，添加进阻塞队列的逻辑删除

        // 提前获取代理对象(事务)
        // 为了让后续线程拿到代理对象，1 可以将他传入阻塞队列，2 将他放到成员变量的位置
        // 创建为了成员变量，在这里进行初始化
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        // 3 返回订单id
        return Result.ok(orderId);
    }

    /***
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 获取用户
        Long userId = UserHolder.getUser().getId();
        // 1. 执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), // 传入空集合
                voucherId.toString(), userId.toString() // 为可变参数
        );
        int r = result.intValue(); // long型转为int

        // 2. 判断结果是否为0
        if (r != 0) {
            // 2.1 不为0，没有购买资格，1，2
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        // 2.2 为0，有购买资格，把下单信息保存到阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        // 订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 用户id
        voucherOrder.setUserId(userId);
        // 优惠券id
        voucherOrder.setVoucherId(voucherId);
        // 放入阻塞队列
        orderTasks.add(voucherOrder);

        // 提前获取代理对象(事务)
        // 为了让后续线程拿到代理对象，1 可以将他传入阻塞队列，2 将他放到成员变量的位置
        // 创建为了成员变量，在这里进行初始化
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        // 3 返回订单id
        return Result.ok(orderId);
    }
    ***/

    /***
    @Override
    // 事务范围为更新数据库范围，减库存操作和创建订单操作，前面的查询不用加事务，封装之后将事务注解加到封装的方法上
//    @Transactional // 牵扯到优惠券库存扣减和订单新增，有两张表操作，加上事务
    public Result seckillVoucher(Long voucherId) {

        // 1.查询优惠券
        // seckillvoucher表和voucher表共享id，秒杀券首先得是优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 2.判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            // 开始时间在当前时间之后，秒杀尚未开始
            return Result.fail("秒杀尚未开始");
        }

        // 3.判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            // 结束时间在当前时间之前，秒杀已经结束
            return Result.fail("秒杀已经结束");
        }

        // 4.判断库存是否充足
        if (voucher.getStock() < 1) {
            // 库存不足
            return Result.fail("库存不足");
        }

        // 总结：
        // 超卖这样的线程安全问题，解决方案有哪些？
        // 1.悲观锁：添加同步锁，让线程串行执行
        // 优点：简单粗暴
        // 缺点：性能一般
        // 2.乐观锁：不加锁，在更新时判断是否有其他线程在修改
        // 优点：性能好
        // 缺点：存在成功率低的问题（库存比较特殊，只要大于0就行；其他情况可以采取分批加锁的方案，比如将100库存分到10张表当中，可以去多张表当中分别去抢，成功率就提高了10倍）

        // 给查询订单，判断订单，新增订单加上悲观锁
        // 选中代码片段 ctrl+alt+m 进行封装

        // 这里还存在事务问题
        // 我们在这里是对createVoucherOrder函数加了事务，没有给seckillVoucher加事务
        // 而seckillVoucher调用createVoucherOrder函数是使用this进行调用的，
        // 而this拿到的是当前VoucherOrderServiceImpl对象，而不是他的代理对象
        // 而事务想要生效是因为spring对当前VoucherOrderServiceImpl这个类进行动态代理，拿到了他的代理对象，用他来去做了事务处理
        // 而现在这个this指的是非代理对象，也就是目标对象，他是没有事务功能的，所以此时spring事务会失效，所以我们需要拿到事务代理的对象
        Long userId = UserHolder.getUser().getId();

        // 集群下的线程并发安全问题
        // idea通过view->tool windows打开service，ctl+d开启另一个服务，修改VM options：-Dserver.port=8082去覆盖yaml文件当中配的端口
        // 然后重启项目就能形成两个节点的集群，然后修改nginx目录下的nginx.conf文件，配置反向代理和负载均衡，
        // 访问nginx时就会反向代理到配置的节点上去，默认使用轮询的负载均衡规则，通过nginx.exe -s reload重新加载配置文件
        // 用同一个用户发起两次请求，发现synchronized无法锁住用户(因为synchronized是节点锁，只对当前节点有效，两台tomcat，两个jvm，两个字符串常量池，两把锁，所以无法锁定)
        // 发现秒杀券库存扣了两个，订单表创建了两个订单，所以在集群环境下，使用synchronized无法锁住
        //
        // 一人一单的并发安全问题
        // 线程1查询订单，线程2查询订单，线程1插入订单，线程2插入订单，由于线程交替执行导致订单被插入两次
        // 后面加入锁，线程1先获取锁，进行订单查询和插入，然后释放锁，线程2在线程1获取锁之后进来，发现获取互斥锁失败，
        // 线程2此时会等待锁释放，然后获取锁，发现订单已经存在，所以不会进行插入，确保了串行执行
        // 在当前jvm内部，锁的原理是由于jvm维护了一个锁的监视器对象，这个监视器对象用的userid，userid在字符串常量池里面，
        // 在这个jvm内部他维护一个池子，所以在id相同的情况下是用一个锁(锁的监视器是同一个)
        // 但是当我们做集群部署的时候，一个新的部署就是一个新的tomcat，也就意味着是一个新的jvm，两个jvm有各自的堆、栈、方法区等，
        // 所以jvm2也有自己的常量池，他用userid作为锁的时候，他的监视器对象又会有一个新的锁监视器，和jvm1的监视器不是同一个
        // 所以新的线程3通过jvm2获取锁的时候，也能获取锁成功
        //
        // 我们的锁监视器在当前jvm的内部可以监视到对应的线程实现互斥，但是有多个jvm，就会有多个锁监视器，每一个jvm内部都会有一个线程是成功的，所以再次出现了线程安全问题

        // 分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁
        // 在集群模式下，synchronized锁失效了，synchronized只能保证单个jvm内部的多个线程之间的互斥，而没有办法让集群模式下的多个jvm进程互斥
        // 我们必须让多个jvm都去使用同一个锁监视器，所以他应该是在jvm外部的多个jvm进程都可以看到的锁监视器
        // 分布式锁特性：
        // 1 多进程可见：只要是jvm外部的基本都可以满足
        // 2 互斥：不管谁来访问都得保证只有一个能拿到锁
        // 3 高可用：大多数情况下来获取锁都得是成功的，不能影响业务
        // 4 高性能：加锁本身就会影响业务的性能，加锁之后会串行执行，如果获取锁很慢则会进一步降低业务性能
        // 5 安全性：考虑异常情况，例：如果获取锁之后在释放之前，锁服务挂了怎么办，会不会产生死锁
        //
        // 不同分布式锁在2345实现上的差异：
        // MySQL
        // 互斥：利用mysql本身的互斥锁机制
        //       mysql或者其他数据库都具备一个事务机制，在执行事务的时候，或者说在执行写操作的时候，mysql会自动分配一个互斥的锁，
        //       所以在多个事务之间就是互斥的，只有一个人能执行，所以可以利用MySQL本身的互斥锁机制实现分布式锁
        // 高可用：好      依赖于mysql本身的可用性，由于MySQL支持主从模式，所以可用性还是不错的
        // 高性能：一般    受限于MySQL的性能，所以性能一般
        // 安全性：好      连接断开，自动释放锁，数据也会回滚
        //
        // redis
        // 互斥：利用setnx这样的互斥命令
        //       setnx是指往redis中set一个数据时，只有当数据不存在时才能set成功，如果已经存在就会set失败，
        // 高可用：好    redis不仅支持主从，还支持集群模式，所以性能非常好
        // 高性能：好
        // 安全性：利用锁超时时间，到期释放，避免服务宕机造成死锁
        //
        // zookeeper
        // 互斥：利用节点的唯一性和有序性实现互斥
        //       利用内部的节点机制，zookeeper内部可以创建数据节点，节点具备唯一性和有序性，还可以创建临时节点
        //       唯一性是创建节点时节点不能重复，有序性是创建节点时节点的id是递增的，可以利用有序性实现互斥，
        //       现在很多线程来zookeeper来创建节点，那么节点的id是单调递增的，约定id最小的那个获取锁成功，
        //       这样就实现互斥了，也可以利用唯一性，都去创建名称一样的节点，只有一个能成功，但一般情况下都是使用有序性
        // 可用性：好      zookeeper也天生支持集群
        // 高性能：一般    zookeeper集群强调强一致性，主从同步会消耗一定时间所以性能是要比redis差的
        // 安全性：好      zookeeper创建的节点往往是临时节点，如果服务宕机，断开连接后节点会自动释放
        //
        // 从安全性考虑，zookeeper和MySQL类似，安全性是比较好的，redis只能利用超时机制去做，但是从可用性和性能去考虑，redis是非常好的

        // redis分布式锁误删问题
        // 线程1获取锁成功，但产生了阻塞，在业务1未完成时触发了超时释放，
        // 线程2此时获取锁执行业务2，但是线程1结束阻塞，完成了业务1，去释放了锁
        // 此时线程3获取了锁执行业务3，此时有两个线程都拿到了锁，再次发生了线程安全问题
        // 造成原因：
        // 线程1业务阻塞导致锁提前释放，此时线程2获取了锁，但是线程1直接释放锁，所以应该在释放锁时进行判断(判断锁的标识是否一致)
        // 改进redis的分布式锁
        // 1. 在获取锁时存入线程标识(可以用UUID标识)
        // 之前线程标识使用的线程id，在jvm内部每创建一个线程，他的数字都会递增，如果在集群模式下，有多个jvm，有可能出现线程id冲突的情况
        // 2. 在释放锁时先获取锁中的线程标识，判断是否与当先线程标识一致：一致则释放锁；不一致则不释放

        // redis分布式锁的原子性问题
        // 线程1获取锁成功，执行业务1，业务1执行完毕，获取锁标识成功，由于判断锁标识和释放锁是两个操作，
        // 在即将释放锁时产生阻塞，由于jvm做full GC时会阻塞所有代码，此时就有可能产生阻塞，
        // 如果阻塞时间足够长就有可能触发锁的超时释放，此时线程2获取锁成功，执行业务2，
        // 但此时线程1阻塞结束，执行释放锁的操作，再次发生误删操作，线程3获取锁成功，再次出现并发安全问题
        // 原因：
        // 判断锁标识和释放锁是两个动作，这两个动作之间产生了阻塞，所以需要保证判断锁标识和释放锁为一个原子性操作
        //
        // redis的lua脚本
        // redis提供了lua脚本功能，在一个脚本中编写多条redis命令，确保多条命令执行时的原子性，
        // redis提供的调用函数：redis.call('命令名称', 'key', '其他参数', ...)
        // redis中的EVAL命令可以执行脚本：EVAL script numkeys key [key ...] arg [arg ...]
        // 例： EVAL "return redis.call('set', 'name', 'jack')" 0
        // 如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，
        // 其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组中获取这些参数(lua中数组角标从1开始)：
        // EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name Rose

        // java调用lua脚本改造分布式锁
        // Lua本身不具备原子性的，是因为redis执行命名是单线程执行的，它会把Lua脚本作为一个命令执行，会阻塞期间接收到其他1线程命令，这就保证了Lua脚本的原子性
        // RedisTemplate调用lua脚本的api(对应EVAL命令):
        // @Override
        // public <T> execute(RedisTemplate<T> script, List<K> keys, Object... args) {
        //     return scriptExecutor.execute(script, keys, args);
        // }

        // 基于redis的分布式锁优化
        // 基于setnx实现的分布式锁存在如下问题：
        // 1 不可重入
        // 同一个线程无法多次获取同一把锁，方法a去调用方法b，在方法a中先去获取锁，然后执行业务调用方法b，
        // b中又要获取同一把锁，如果是不可重入的，此时方法b无法获取，需要等待锁的释放，而方法a没执行结束，无法释放锁，就出现了死锁
        // 2 不可重试
        // 获取锁只尝试一次就返回false，没有重试机制，有些业务下需要锁是阻塞的，或者说是可重试的
        // 3 超时释放
        // 锁超时释放虽然可以避免死锁，然是如果业务执行耗时比较长，也会导致锁释放，存在安全隐患
        // 4 主从一致性
        // 读写分离模式：执行写操作时访问主节点，执行读操作时访问从节点，可以在多个从节点上完成读的操作，提高服务的并发能力和高可用性，主节点宕机还可以从从节点选出新的主节点
        // 如果redis提供了主从集群，主从同步存在延迟，当主宕机时，如果从并未同步主中的锁数据，则会出现锁失效
        // redisson介绍
        // redisson是一个在redis的基础上实现的java驻内存数据网格(In-Memory Data Grid)。他不仅提供了一系列的分布式的java常用对象，还提供了许多的分布式服务，其中就包含了各种分布式锁的实现。
        // 分布式锁(Lock)和同步器(Synchronizer)：可重入锁(Reentrant Lock)、公平锁(Fair Lock)、联锁(MultiLock)、红锁(RedLock)、
        //                                      读写锁(ReadWriteLock)、信号量(Semaphore)、可过期信号量(PermitExpirableSemaphore)、闭锁(CountDownLatch)

        // redisson
        // 1 引入依赖
        // <!--引入redisson依赖-->
        //        <dependency>
        //            <groupId>org.redisson</groupId>
        //            <artifactId>redisson</artifactId>
        //            <version>3.13.6</version>
        //        </dependency>
        // 2 配置redisson客户端
        // config/RedisConfig
        // 3 使用redis分布式锁
        // test/RedissonTest
        //
        // redisson可重入锁原理
        // 在获取锁的时候，在判断已经有人获取锁的情况下，去判断获取锁的是不是自己，是不是同一个线程，
        // 如果是同一个线程，也会让他获取锁，用一个计数器去记录重入的次数，每一次重入，计数器+1，释放锁的时候计数器-1
        // 在获取锁的时候不仅要记录线程，还需要记录重入的次数，此时基于string的锁结构就不行了，应该采取hash类型
        // 在key的位置记录锁的名称，在value位置记录线程，在field位置记录重入次数
        // hash结构锁流程：
        // 开始
        // 判断锁是否存在，否，获取锁并添加线程标识，设置锁过期时间，
        // 是，判断锁标识是否是自己，否，获取锁失败，是，锁计数器+1，设置锁过期时间
        // 执行业务，判断锁是否是自己，否，锁已释放(超时)，
        // 是，锁计数器-1，判断锁计数器是否为0，是，释放锁，否，重置锁有效期，执行业务
        // 无论是获取锁还是释放锁，都比string结构的流程复杂，具有多个步骤，一定要采取lua脚本确保获取锁和释放锁的原子性

        // redis秒杀优化
        // 秒杀流程，查询优惠券->判断秒杀库存->查询订单->校验一人一单->减库存->创建订单
        // 查询优惠券、查询订单，减库存，创建订单都会操作数据库，而数据库的并发能力比较差
        // 后面两个是写操作，而且还加了分布式锁，导致业务整体耗时较长，1000个用户秒杀200个优惠券达到了200多毫秒，
        // 对于秒杀资格的判断耗时比较少，可以将其进行拆分，将优惠券和订单信息缓存到redis中，
        // 将对秒杀资格的判断操作放到redis中去做，主线程进来后去redis中进行秒杀资格的判断，如果发现有资格再去进行减库存和下单的操作，
        // 减库存和下单操作开启独立线程去执行，独立线程需要知道订单信息才能去执行，所以需要主线程保存优惠券id、用户id、订单id到阻塞队列，返回订单号给用户(服务员->小票->后厨，用户取餐号)，
        // 将来就可以开启独立线程去读取队列里面的信息，去完成减库存和下单，
        // 给用户返回订单id一方面用户可以拿着订单id去孵卵，另一方面可以代表用户抢单成功，
        // 虽然此时订单还未创建，在阻塞队列里面，但是会确保他会创建，将来异步地开启一个独立线程去读取队列中的信息，完成下单操作，
        // 这样一来，接收用户请求后的流程就变成了：进来直接判断用户的秒杀资格，而且判断是在redis里面做的，做完以后只需要将他保存到队列里面就结束了，
        // 整个业务流程就变短了，而且基于redis，整个业务的性能可以得到极大地提升，吞吐能力和并发能力大大提高，
        //
        // 如何在redis里面完成对于秒杀库存的判断和一人一单的判断
        // 库存只需要一个简单的string结构，key为优惠券id，value为库存的值，判断库存只需要判断value是否>0，
        // 判断用户有购买资格之后，需要在redis里面将库存-1，需要在redis里面提前预减库存，
        // 实现一人一单需要在redis里面当前优惠券被哪些人购买过，再有用户过来只需要判单是否存在，
        // 如果存在，证明用户购买过，则不能再次购买，
        // 一个优惠券库存有很多，所以购买的用户也有很多，所以选取的结构需要在一个key里面保存很多值，也就是一个list列表，
        // 一人一单，所以在这些优惠券里面保存的用户id是不能重复的，需要保证唯一性，所以需要选取set集合来存储，
        // 库存:string：
        // key：stock:vid:7 value：100
        // 订单:set：
        // key：order:vid:7 value：1， 2， 3， 5， 7， 8(用户id)
        // 业务流程:
        // 开始，判断库存是否充足，否，返回1，结束
        // 是，判断用户是否已经下单，是，返回2，结束
        // 否，扣减库存，将userId存入当前优惠券的set集合，返回0，结束
        // 还需要确保业务流程执行的原子性，所以需要将业务流程写在lua脚本里面
        // java代码流程:
        // 开始，执行lua脚本，判断结果是否为0，否，返回异常信息，结束
        // 是，将优惠券id，用户id和订单id存入阻塞队列，返回订单id，结束
        // 将来只需要开启独立线程来读取阻塞队列异步写入数据库就行，
        // 在返回订单id给用户的那一刻秒杀业务已经结束，用户已经可以拿着订单id去付款了，
        // 所以我们什么时候将优惠券信息、用户信息写入数据库进行下单减库存的操作就没有那么重要了，
        // 时效性上要求就没有那么高了，完全可以按照数据库可以承受的频率去将数据写入数据库，
        // 当然，想要提高写入数据库的性能，可以多开几个线程，甚至可以将单个的写改成批量的写，从而提高异步操作的效率

        // 改进秒杀业务，提高并发性能
        // key：stock:vid:7 value：100
        // 订单:set：
        // key：order:vid:7 value：1， 2， 3， 5， 7， 8(用户id)
        // 在redis里面进行秒杀资格判断第一个要判断的就是库存，所以需要:
        // 1 在新增秒杀优惠券的同时，将优惠券信息保存到redis当中
        // 在VoucherServiceImpl增加将秒杀券保存到redis当中的操作
        // 有关购买这个优惠券的用户信息是在秒杀的过程中添加到的，所以先不需要考虑
        // 完成1后，需要进行秒杀资格判断：
        // 并发问题:(因为redis执行命令是单线程，然后lua脚本保证了多条命令的原子性，所以就不会有多个线程交叉执行redis命令的情况)
        // 2 基于lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
        // 编写seckill.lua，通过java执行lua脚本
        // 3 判断抢购是否成功，如果成功，将优惠券id和用户id封装好后放入阻塞队列
        // 4 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

//        synchronized (userId.toString().intern()) { // 下面：通过redis来加锁，解决多节点线程安全问题
        // 创建锁对象
        String lockName = "order:" + userId;
        SimpleRedisLock lock = new SimpleRedisLock(lockName, stringRedisTemplate);
        RLock lock1 = redissonClient.getLock("lock:order:" + userId);
        // 获取锁
        boolean isLock = lock1.tryLock();
        // 判断是否获取锁成功
        if (!isLock) {
            // 获取失败，返回错误或重试
            return Result.fail("一个人只允许下一单");
        }
        // ctrl+alt+t
        try {
            // 利用AopContext.currentProxy()方法就能够拿到当前对象的代理对象了，当前对象是IVoucherOrderService接口，所以是他的代理对象
            // 我们用代理对象去调用createVoucherOrder函数，而不是this，这样就会由spring去进行管理，因为代理对象是由spring创建的，所以他具有事务
            // 获取代理对象(事务)
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
            // return createVoucherOrder(voucherId);
            // 上行等价于 return this.createVoucherOrder(voucherId);
        } finally {
            // 释放锁
            lock1.unlock();
        }
//        }
    }
     ***/

    /***
    @Transactional // 事务范围为减库存和创建订单
    // 不建议把synchronized加到方法上，加到方法上锁的范围就变成整个方法，锁的对象是this，
    // 意味着任何一个用户来了都需要加这个锁，而且是同一把锁，意味着整个方法就是串行执行了，性能就会很差
    // 而在一人一单的场景下，同一个用户来了才去判断并发安全问题，如果不是同一个用户，不需要去加锁
    // 所以这里加的锁不应该是service，而是当前用户，所以应该给用户id加锁，所以应该把锁的范围缩小
    // 同一个用户加用一把锁，不同用户加不同锁，将锁定资源范围减小
//    public synchronized Result createVoucherOrder(Long voucherId) { // 同步锁为this，当前对象，线程安全
    public Result createVoucherOrder(Long voucherId) {
        // 4.5 一人一单
        // 使用jmeter测试200线程抢购100库存，失败率95%，出现10个订单，仍然存在一人多单，需要加锁，加乐观锁需要判断库存是否修改，这里是新增订单，无法判断，所以用悲观锁
        Long userId = UserHolder.getUser().getId();

        // 不用synchronized方法，使用synchronized关键字
        // 我们期望id值一样的作为一把锁，每一个请求来，id对象都是一个全新的id对象，因此对象变了锁就变了
        // 我们要求值一样，所以这里用了toString，toString方法里面底层调用的是Long.java的一个静态的toString函数，在他的内部new了一个字符串
        // 所以每调一次toString也是一个全新的字符串对象，所以锁对象还是在变
        // 调用字符串的intern方法，这个方法返回字符串对象的规范表示，去字符串常量池里找到和你值一样的字符串地址返回
        // 如果id的值是10，不管new了多少个字符串，只要值是一样的，最后返回的结果都是一样的，这样可以确保当用户id的值一样锁就一样
        // 这样就可以锁当前用户，而不同的用户就不会被锁定，这样锁定范围变小，性能就得到了很大的提升
        //
        // 但是在方法内部加锁，比如现在开启事务开始执行，执行之后获取锁，然后进行查询，然后减库存，提交订单，然后先释放锁，之后才会提交事务
        // 因为我们的事务是交给spring管理的，他是在我们函数执行完以后由spring做的提交，这时候锁在synchronized{}大括号执行完以后已经释放了
        // 锁释放后意味着其他线程可以进来，而此时事务尚未提交，如果有其他线程进来查询订单，但是新增的订单很有可能还没有写入数据库，因为还没有提交
        // 所以查询的时候依然不存在订单，有可能出现并发安全问题，因此锁定的范围需要增大，应该把整个函数锁起来，应该等事务提交时候再去释放锁
//        synchronized (userId.toString().intern()) {
            // 4.1 查询订单
            int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
            // 4.2 判断是否存在
            if (count > 0) {
                return Result.fail("用户已经购买过一次");
            }

            // 5.扣减库存
            // cas实现乐观锁
            // 加乐观锁之前，jmeter测试200线程进行抢购100库存，失败率45%，出现超卖5%(50-45)，数据库优惠券库存变成-10
            // 加乐观锁(cas实现)之后，jmeter测试200线程抢购100库存，失败率增加，只生成20个订单，失败率90% (where id = ? and stock = ?)
            // 修改判断条件，只要库存大于0都可以进行修改 (where id = ? and stock > 0)，失败率50%，符合需求200抢购100，失败一半
            boolean sucess = seckillVoucherService.update()
                    .setSql("stock = stock - 1") // set stock = stock - 1
                    .eq("voucher_id", voucherId).gt("stock", 0).update(); // where id = ? and stock > 0
            //.eq("voucher_id", voucherId).eq("stock", voucher.getStock()).update(); // where id = ? and stock = ?

            if (!sucess) {
                // 扣减失败
                return Result.fail("库存不足");
            }

            // 6.创建订单
            VoucherOrder voucherOrder = new VoucherOrder();
            // 6.1 订单id
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
            // 6.2 用户id
            voucherOrder.setUserId(userId);
            // 6.3 代金券id
            voucherOrder.setVoucherId(voucherId);
            // 6.4 将订单写入数据库
            save(voucherOrder);

            // 7.返回订单id
            log.debug("抢购成功");
            return Result.ok(orderId);
//        }
    }
     ***/

    @Transactional
    @Override
    public void crtVoucherOrder(VoucherOrder voucherOrder) {
        // 一人一单
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        // 查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if (count > 0) {
            log.error("用户已经买过一次");
            return;
        }
        // 扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
                .update();
        if (!success) {
            log.error("库存不足");
            return;
        }
        // 创建订单
        save(voucherOrder);
    }
}